package com.rn.gypsophila.grpc;

import com.google.common.collect.Lists;
import io.grpc.Context;
import io.grpc.examples.calculate.CalculateProto;
import io.grpc.examples.calculate.CalculateServiceGrpc.CalculateServiceImplBase;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class CalculateServiceImpl extends CalculateServiceImplBase {
    public static final ReentrantLock LOCK = new ReentrantLock();
    public static Condition condition = LOCK.newCondition();

    @Override
    public StreamObserver<CalculateProto.Value> getResult(StreamObserver<CalculateProto.Result> responseObserver) {
        return new StreamObserver<CalculateProto.Value>() {
            @Override
            public void onNext(CalculateProto.Value value) {
                log.info("接收到消息为:{}", value.getValue());
                List<String> txHashList = Lists.newArrayList();
                txHashList.add("tt");
                txHashList.add("cc");
                CalculateProto.Result response = CalculateProto.Result.newBuilder().addAllTxHash(txHashList).build();
                while (!Context.current().isCancelled()) {
                    LOCK.lock();
                    try {
                        condition.await();
                        responseObserver.onNext(response);
                        log.info("返回消息为:{}", response);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        LOCK.unlock();
                    }
                }
            }

            @Override
            public void onError(Throwable throwable) {
                log.warn("调用出错", throwable);
            }

            @Override
            public void onCompleted() {
                log.warn("调用完成！");
                responseObserver.onCompleted();
            }
        };
    }
}